package spike.pull;

import java.util.Date;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;

public class ConnectionPoolImpl<T> implements ConnectionPool<T> {
    private ConnectionSource<T> src;
    private int init;
    private long keepAliveTime;
    private TimeUnit unit;
    private Semaphore semaphore;
    private ConcurrentLinkedDeque<TemporaryConnection> connections = new ConcurrentLinkedDeque<>();
    volatile private boolean isStopped = false;

    public ConnectionPoolImpl(ConnectionSource<T> src, int init, int max, long keepAliveTime, TimeUnit unit) {
        this.src = src;
        this.init = init;
        this.keepAliveTime = keepAliveTime;
        this.unit = unit;
        semaphore = new Semaphore(max, false);
    }

    public ConcurrentLinkedDeque<TemporaryConnection> getConnections() {
        return connections;
    }

    public void start() {
        for (int i = 0; i < init; i++) {
            TemporaryConnection tc = new TemporaryConnection(src.open());
            connections.add(tc);
        }

        if (keepAliveTime > 0){
            Executors.newScheduledThreadPool(1).scheduleAtFixedRate(() -> {
                for (TemporaryConnection tc : connections) {
                    try {
                        if (connections.size() <= init) {
                            break;
                        }

                        Date date = new Date();
                        long liveTime = date.getTime() - tc.getTimeCreation().getTime();
                        if (liveTime > TimeUnit.MILLISECONDS.convert(keepAliveTime, unit)) {
                            if (connections.remove(tc)) {
                                src.close(tc.getConnection());
                            }
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }, keepAliveTime, keepAliveTime, unit);
        }
    }

    public void stop() throws InterruptedException {
        isStopped = true;
        while(semaphore.hasQueuedThreads()){
            semaphore.release();
        }
    }

    @Override
    public T get() throws InterruptedException {
        if (isStopped)
            return null;

        semaphore.acquire();
        return getFreeConnection();
    }

    @Override
    public T tryGet() {
        if (isStopped)
            return null;

        if (semaphore.tryAcquire()) {
            return getFreeConnection();
        }
        return null;
    }

    @Override
    public T tryGet(long time, TimeUnit unit) throws InterruptedException {
        if (isStopped)
            return null;

        if (semaphore.tryAcquire(time, unit)) {
            return getFreeConnection();
        }
        return null;
    }

    private T getFreeConnection() {
        if (isStopped)
            return null;

        if (connections.isEmpty()) {
            return src.open();
        }
        return connections.remove().getConnection();
    }

    @Override
    public void release(T object) {
        connections.add( new TemporaryConnection(object));
        semaphore.release();
    }

    public boolean isStopped() {
        return isStopped;
    }

    private class TemporaryConnection{
        private T connection;
        private Date timeCreation;

        public TemporaryConnection(T connection) {
            this.timeCreation = new Date();
            this.connection = connection;
        }

        public T getConnection() {
            return connection;
        }

        public Date getTimeCreation() {
            return timeCreation;
        }

    }
}
